package controller

import (
	"encoding/json"
	"errors"
	"recorder/control"
	"recorder/internal"
	"recorder/log"
	"sync/atomic"
	"time"
)

var currentWorkerCount int32 = 0

const (
	ackTimeout = 3 * time.Second
)

type Worker struct {
	conn *control.Conn
	done chan struct{}
}

func getCurrentWorkerCount() int32 {
	return atomic.LoadInt32(&currentWorkerCount)
}

func NewWorker(conn *control.Conn) *Worker {
	atomic.AddInt32(&currentWorkerCount, 1)
	return &Worker{
		conn: conn,
		done: make(chan struct{}),
	}
}

func (w *Worker) Id() string {
	return w.conn.Id()
}

func (w *Worker) Close() {
	atomic.AddInt32(&currentWorkerCount, -1)
	close(w.done)
	w.conn.Close()
}

func (w *Worker) RunJob(j *internal.Job) (*internal.Result, error) {

	log.Info("worker %s: sending job %v", w.Id(), j)

	json, _ := json.Marshal(j)
	log.Info("RunJob.j === %s", string(json))

	beginTime := time.Now()
	m := &control.Message{
		Action: control.ActionRun,
		Job:    j,
	}
	select {
	case <-w.conn.Dis:
		log.Warning("worker %s: disconnected while sending job %v", w.Id(), j)
		return nil, errors.New("disconnected")
	case w.conn.Out <- m:
	}
	log.Info("worker %s: job %v sent", w.Id(), j)

	ackTimer := time.NewTimer(ackTimeout)
	defer ackTimer.Stop()
	select {
	case <-ackTimer.C:
		log.Warning("worker %s: wait for ack timeout", w.Id())
		return nil, errors.New("timeout")
	case <-w.done:
		log.Warning("worker %s: closed while waiting for ack", w.Id())
		return nil, errors.New("aborted")
	case <-w.conn.Dis:
		log.Warning("worker %s: disconnected while waiting for ack", w.Id())
		return nil, errors.New("disconnected")
	case m := <-w.conn.In:
		if m == nil || m.Action != control.ActionAck {
			log.Warning("worker %s: waiting for ack, got %+v", w.Id(), m)
			return nil, errors.New("unexpected")
		}
	}
	log.Info("worker %s: recv ack", w.Id())
	r := internal.Result{Job: j.Scene}

	select {
	case <-w.done:
		log.Warning("worker %s: closed while waiting for done", w.Id())
		return nil, errors.New("aborted")
	case <-w.conn.Dis:
		log.Warning("worker %s: disconnected while waiting for done", w.Id())
		return nil, errors.New("disconnected")
	case m := <-w.conn.In:
		if m == nil || m.Action != control.ActionDone {
			log.Warning("worker %s: waiting for done, got %+v", w.Id(), m)
			return nil, errors.New("unexpected")
		}
		r.Error = m.Result.Error
	}
	costTime := time.Now().Sub(beginTime)
	log.Info("worker %s: recv done for job %s, total cost %v", w.Id(), j.Scene, costTime)

	select {
	case <-w.done:
		log.Error("worker %s: closed while sending ack", w.Id())
		return nil, errors.New("aborted")
	case <-w.conn.Dis:
		log.Error("worker %s: disconnected while sending ack", w.Id())
		return nil, errors.New("disconnected")
	case w.conn.Out <- &control.Message{Action: control.ActionAck}:
	}
	log.Info("worker %s: ack sent", w.Id())

	return &r, nil
}
